import { AppTask } from "../../app.task.class";
import { BehaviorSubject, Observable } from "rxjs";
import mqtt from "mqtt";
import { ToSRMqttConverter } from "../converter/sr.converter";
import { SrAlertMqttPublishTask } from "./sr.alert.mqtt.publish.task";
import { V3proSubjectMessage } from "../../v3pro-task/model/v3pro.model";

/**
 * 这里收到的消息已经通过 topic 分割过了
 */
export class SRAlertTopicHandlerTask extends AppTask {
  // mqtt client
  private __mqttClient: any;

  /**
   * 构造函数
   * @param __mqttSubject
   * @param __mqttMessageSubject
   */
  constructor(
    private __mqttSubject: BehaviorSubject<mqtt.MqttClient | undefined>,
    private __mqttMessageSubject: Observable<V3proSubjectMessage>,
  ) {
    super();
    const mqttDescription = this.__mqttSubject.subscribe(this.__mqttEventHandler.bind(this));
    const messageDescription = this.__mqttMessageSubject.subscribe(
      this.__mqttMessageEventHandler.bind(this),
    );
    this.onStop = () => {
      mqttDescription.unsubscribe();
      messageDescription.unsubscribe();
    };
  }

  /**
   * mqtt client 事件处理
   * @param mqttClient
   */
  async __mqttEventHandler(mqttClient: mqtt.MqttClient | undefined) {
    this.__mqttClient = mqttClient;
    if (mqttClient) {
      await this.start();
      this.__lastMQTTTask?.start().then();
    } else await this.pause();
  }

  // 最近的一次任务
  private __lastMQTTTask: AppTask | undefined;

  /**
   * mqtt消息事件处理
   * @param message
   */
  async __mqttMessageEventHandler(message: V3proSubjectMessage | undefined) {
    if (!message) return;
    if (this.__lastMQTTTask) await this.__lastMQTTTask.stop();
    this.__lastMQTTTask = new SrAlertMqttPublishTask(
      this.__mqttClient,
      message.srMqttTopic!,
      ToSRMqttConverter.fromV3proSubjectMessage(message),
    );
    if (!this.__mqttClient) return;
    this.__lastMQTTTask.start().then();
    this.__lastMQTTTask = undefined;
  }

  __pause(): Promise<boolean> {
    return Promise.resolve(false);
  }

  __restore(): Promise<boolean> {
    return Promise.resolve(false);
  }

  async __run(): Promise<void> {
    await this.__await();
  }

  __stop(): Promise<boolean> {
    return Promise.resolve(false);
  }
}
